package ru.zombator.taskman.server.query.udp;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.DatagramChannel;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import ru.zombator.taskman.core.query.IPacket;
import ru.zombator.taskman.core.utils.PacketUtils;
import ru.zombator.taskman.server.query.BaseServer;

/**
 * UDP-сервер. Принимает дейтаграммы последовательно, а отвечает на них
 * параллельно (в отдельном потоке).
 * 
 * @author nvamelichev
 *
 */
public class UdpServer extends BaseServer {
	private static final long serialVersionUID = 1L;
	
	private final int UDP_SERVER_PORT = 5555;
	
	/**
	 * размер дейтаграммы - 20 килобайт
	 */
	public final int PACKET_SIZE = 1024 * 20;
	
	private DatagramChannel srvSocketChannel;

	private final ExecutorService responders = Executors.newCachedThreadPool();
	private final ExecutorService listener = Executors.newFixedThreadPool(1);
	private AtomicBoolean isRunning = new AtomicBoolean();
	
	private class ListenerThread implements Callable<Void> {
		@Override
		public Void call() {
			// ждем соединения
			while (isRunning.get()) {
				ByteBuffer dstBuf = ByteBuffer.allocate(PACKET_SIZE);
				dstBuf.order(ByteOrder.LITTLE_ENDIAN);
				
				SocketAddress addr = null;
				try {
					addr = srvSocketChannel.receive(dstBuf);
				} catch (IOException e) {
					e.printStackTrace();
				}
				ResponderThread respThread = new ResponderThread();
				respThread.setBuffer(dstBuf);
				respThread.setCh(srvSocketChannel);
				respThread.setAddr(addr);
				responders.submit(respThread);
			}
			try {
				srvSocketChannel.close();
			} catch (IOException e) {
				e.printStackTrace();
			}			
			responders.shutdownNow();
			return null;
		}
	}
	
	private class ResponderThread implements Callable<Void> {
		private ByteBuffer buf;
		private DatagramChannel ch;
		private SocketAddress addr;
		
		public void setAddr(SocketAddress addr) {
			this.addr = addr;
		}
		
		public void setCh(DatagramChannel ch) {
			this.ch = ch;
		}
		
		public void setBuffer(ByteBuffer buf) {
			this.buf = buf;
		}
		
		@Override
		public Void call() throws Exception {
			buf.rewind();
			int payloadSize = buf.getInt();
			IPacket pktRequest = PacketUtils.bufferToPacket(buf, payloadSize);
			
			String ipAddress = ((InetSocketAddress) addr).getAddress().getHostAddress();
			IPacket pktResponse = getResponse(pktRequest, ipAddress);
			
			ByteBuffer outBuf = PacketUtils.packetToBuffer(pktResponse, PACKET_SIZE);
			ch.send(outBuf, addr);
			
			return null;
		}
	}
	
	public UdpServer() throws IOException {
		srvSocketChannel = DatagramChannel.open();
		srvSocketChannel.socket().bind(new InetSocketAddress(UDP_SERVER_PORT));
		srvSocketChannel.configureBlocking(true);
		
		isRunning.set(true);
		listener.submit(new ListenerThread());
	}

	@Override
	public void dispose() {
		isRunning.set(false);
		responders.shutdown();
		listener.shutdown();
	}
}
